//
//  dataoperations.cc
//  Draco
//
//  Created by Liu Dafan on 12-2-16.
//  Copyright (c) 2012年 __MyCompanyName__. All rights reserved.
//

#include <iostream>

#include "error_code.h"
#include "hbase_client.h"
#include "hdfs_client.h"
#include "datastream.h"
#include "data_operations.h"
#include "data_info_transfer.h"

using namespace std;

const char*  DataOperations::DATA_FILE_PREFIX = "data_file";
const char*  DataOperations::DIRID_FILE = "dirid";
ConnPool*    DataOperations::_cp;
char         DataOperations::_rootdir[256];

static RowResult* GenerateRowResult(const DIR_INFO& dirinfo)
{
    uint64_t timestamp = 0;
    char buf[1024];
    RowResult* pRow = new RowResult();

    sprintf(buf, DIR_INFO_STRING_TAG[DIR_INFO_ID], dirinfo._dirID._id);
    pRow->insert(DIR_INFO_STRING_STR[DIR_INFO_ID], buf, timestamp);

    pRow->insert(DIR_INFO_STRING_STR[DIR_INFO_NAME], dirinfo._name, timestamp);

    sprintf(buf, DIR_INFO_STRING_TAG[DIR_INFO_PARENT_ID], dirinfo._parentID._id);
    pRow->insert(DIR_INFO_STRING_STR[DIR_INFO_PARENT_ID], buf, timestamp);

    sprintf(buf, DIR_INFO_STRING_TAG[DIR_INFO_COUNT], dirinfo._fileCount);
    pRow->insert(DIR_INFO_STRING_STR[DIR_INFO_COUNT], buf, timestamp);

    sprintf(buf, DIR_INFO_STRING_TAG[DIR_INFO_SIZE], dirinfo._totalSize);
    pRow->insert(DIR_INFO_STRING_STR[DIR_INFO_SIZE], buf, timestamp);

    sprintf(buf, DIR_INFO_STRING_TAG[DIR_INFO_CREATE_TIME], dirinfo._createTime);
    pRow->insert(DIR_INFO_STRING_STR[DIR_INFO_CREATE_TIME], buf, timestamp);

    sprintf(buf, DIR_INFO_STRING_TAG[DIR_INFO_UPDATE_TIME], dirinfo._updateTime);
    pRow->insert(DIR_INFO_STRING_STR[DIR_INFO_UPDATE_TIME], buf, timestamp);
    
    sprintf(buf, DIR_INFO_STRING_TAG[DIR_INFO_DELETE_TIME], dirinfo._deleteTime);
    pRow->insert(DIR_INFO_STRING_STR[DIR_INFO_DELETE_TIME], buf, timestamp);
    
    sprintf(buf, DIR_INFO_STRING_TAG[DIR_INFO_IS_DELETE], dirinfo._isDeleted);
    pRow->insert(DIR_INFO_STRING_STR[DIR_INFO_IS_DELETE], buf, timestamp);
    
    pRow->next = NULL;
    
    return pRow;
}

static RowResult* GenerateRowResult(const FILE_INFO& meta)
{
    uint64_t timestamp = 0;
    char buf[1024];
    RowResult* pRow = new RowResult();

    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_NAME], meta._name, timestamp);
    
    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_VERSION], meta._fileVersion);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_VERSION], buf, timestamp);
    
    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_PARENT_ID], meta._parentID._id);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_PARENT_ID], buf, timestamp);
        
    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_CREATE_TIME], meta._createTime);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_CREATE_TIME], buf, timestamp);
    
    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_DELETE_TIME], meta._deleteTime);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_DELETE_TIME], buf, timestamp);

    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_LENGTH], meta._length);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_LENGTH], buf, timestamp);

    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_COMPRESS], meta._compressType);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_COMPRESS], buf, timestamp);
    
    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_SECURITY], meta._securityType);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_SECURITY], buf, timestamp);
    
    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_IS_DELETE], meta._isDeleted);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_IS_DELETE], buf, timestamp);
    
    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_IS_SHARING], meta._isSharing);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_IS_SHARING], buf, timestamp);

    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_DATA_CRC], meta._dataCRC);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_DATA_CRC], buf, timestamp);
    
    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_HEAD_CRC], meta._headCRC);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_HEAD_CRC], buf, timestamp);

    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_LOC_FILE], meta._loc._fileName, timestamp);

    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_LOC_OFFSET], meta._loc._offset);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_LOC_OFFSET], buf, timestamp);

    sprintf(buf, FILE_INFO_STRING_TAG[FILE_INFO_LOC_LENGTH], meta._loc._length);
    pRow->insert(FILE_INFO_STRING_STR[FILE_INFO_LOC_LENGTH], buf, timestamp);

    pRow->next = NULL;
    
    return pRow;
}

static HResult ReadFromRowResult(RowResult* result, DIR_INFO& dirinfo)
{
    if (result == NULL)
    {
        return ERROR_CODE_ILLEGAL_ARGUMENT;
    }

    const char* buf = NULL;
    int res;
    
    ColumnValue *cv;
    map<string, string> items;
    while ((cv = result->nextcolumn()) != NULL)
    {
        items[cv->column] = cv->value;
    }
    
    buf = items[DIR_INFO_STRING_STR[DIR_INFO_ID]].c_str();
    res = sscanf(buf, DIR_INFO_STRING_TAG[DIR_INFO_ID], &dirinfo._dirID._id);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }

    strcpy(dirinfo._name, items[DIR_INFO_STRING_STR[DIR_INFO_NAME]].c_str());
    
    buf = items[DIR_INFO_STRING_STR[DIR_INFO_PARENT_ID]].c_str();
    res = sscanf(buf, DIR_INFO_STRING_TAG[DIR_INFO_PARENT_ID], &dirinfo._parentID);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[DIR_INFO_STRING_STR[DIR_INFO_COUNT]].c_str();
    res = sscanf(buf, DIR_INFO_STRING_TAG[DIR_INFO_COUNT], &dirinfo._fileCount);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }

    buf = items[DIR_INFO_STRING_STR[DIR_INFO_SIZE]].c_str();
    res = sscanf(buf, DIR_INFO_STRING_TAG[DIR_INFO_SIZE], &dirinfo._totalSize);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[DIR_INFO_STRING_STR[DIR_INFO_CREATE_TIME]].c_str();
    res = sscanf(buf, DIR_INFO_STRING_TAG[DIR_INFO_CREATE_TIME], &dirinfo._createTime);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[DIR_INFO_STRING_STR[DIR_INFO_UPDATE_TIME]].c_str();
    res = sscanf(buf, DIR_INFO_STRING_TAG[DIR_INFO_UPDATE_TIME], &dirinfo._updateTime);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[DIR_INFO_STRING_STR[DIR_INFO_DELETE_TIME]].c_str();
    res = sscanf(buf, DIR_INFO_STRING_TAG[DIR_INFO_DELETE_TIME], &dirinfo._deleteTime);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[DIR_INFO_STRING_STR[DIR_INFO_IS_DELETE]].c_str();
    res = sscanf(buf, DIR_INFO_STRING_TAG[DIR_INFO_IS_DELETE], &dirinfo._isDeleted);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    return ERROR_CODE_OK;
}

static HResult ReadFromRowResult(RowResult* result, FILE_INFO& fileinfo)
{
    if (result == NULL)
    {
        return ERROR_CODE_ILLEGAL_ARGUMENT;
    }
    
    const char* buf = NULL;
    int res;
    
    ColumnValue *cv;
    map<string, string> items;
    while ((cv = result->nextcolumn()) != NULL)
    {
        items[cv->column] = cv->value;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_PARENT_ID]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_PARENT_ID], &fileinfo._parentID._id);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    strcpy(fileinfo._name, items[FILE_INFO_STRING_STR[FILE_INFO_NAME]].c_str());
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_VERSION]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_VERSION], &fileinfo._fileVersion);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_CREATE_TIME]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_CREATE_TIME], &fileinfo._createTime);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_DELETE_TIME]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_DELETE_TIME], &fileinfo._deleteTime);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_LENGTH]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_LENGTH], &fileinfo._length);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_COMPRESS]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_COMPRESS], &fileinfo._compressType);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_SECURITY]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_SECURITY], &fileinfo._securityType);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_IS_DELETE]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_IS_DELETE], &fileinfo._isDeleted);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_IS_SHARING]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_IS_SHARING], &fileinfo._isSharing);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_DATA_CRC]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_DATA_CRC], &fileinfo._dataCRC);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_HEAD_CRC]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_HEAD_CRC], &fileinfo._headCRC);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_LOC_FILE]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_LOC_FILE], &fileinfo._loc._fileName);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_LOC_OFFSET]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_LOC_OFFSET], &fileinfo._loc._offset);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    buf = items[FILE_INFO_STRING_STR[FILE_INFO_LOC_LENGTH]].c_str();
    res = sscanf(buf, FILE_INFO_STRING_TAG[FILE_INFO_LOC_LENGTH], &fileinfo._loc._length);
    if (res != 1)
    {
        DERROR("Parse error: %s", buf);
        return ERROR_CODE_INTERNAL;
    }
    
    return ERROR_CODE_OK;
}

void DataOperations::Init(const Config* conf)
{
    _cp = init_connpool(conf);
    strcpy(DataOperations::_rootdir, conf->rootdir);
}

void DataOperations::UnInit()
{
    close_connpool(_cp);
}

HResult DataOperations::WriteDirectoryInfo(const std::string& tablename, const DIR_INFO& dirinfo)
{
    HResult ret = ERROR_CODE_OK;
    
    // generate RowResult from Directory info
    string rowkey = dirinfo.GetKey();
    
    RowResult* row = GenerateRowResult(dirinfo);
    
    // get connection
    hbase_conn *hconn = (hbase_conn *)_cp->get(CONN_POOL_TYPE_HBASE);
	HbaseOp *conn = hconn->conn;
    
    int trytimes = 0;
    
	if (hconn == NULL) {
		DERROR("WriteDirectoryInfo error: hbase server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        goto done;
	}
    
    //put lines in hbase
	while (1) {
		ret = conn->put(tablename.c_str(), rowkey.c_str(), row);
		if (ret < 0) {
			if (conn->errnum == ERROR_CODE_IOERROR) {
				DERROR("hbase put table <%s> error, io error, ignore the request.\n", tablename.c_str());
				ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
				goto done;
			} else if (conn->errnum == ERROR_CODE_SOCKET_CLOSED) {
				trytimes += 1;
				set_inactive(hconn);
				if (trytimes > 3) {
					DERROR("hbase put table <%s> try %d times, ignore the request.\n", tablename.c_str(), trytimes-1);
					ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
					goto done;
				}
				DERROR("hbase put table <%s> error, socket broken, try again(%d).\n", tablename.c_str(), trytimes);
				//hconn->pool->active = 0;
				_cp->put(CONN_POOL_TYPE_HBASE, hconn);
				hconn = (struct hbase_conn *)_cp->get(CONN_POOL_TYPE_HBASE);
				if (hconn == NULL) {
					DERROR("hbase put table <%s> error, server resource unavailable.\n", tablename.c_str());
					ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
					goto done;
				}
				conn = hconn->conn;
				continue;
			} else if (conn->errnum == ERROR_CODE_ILLEGAL_ARGUMENT) {
				DERROR("hbase put table <%s> error, invalid parameter.\n", tablename.c_str());
				ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
				goto done;
			} else if (conn->errnum == ERROR_CODE_DATA_STATUS_UNKNOWN) {
				DERROR("hbase put table <%s> error, data status unknown.\n", tablename.c_str());
				ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
				goto done;
			}
		}
		break;
	}
    
done:
    free_rows(row);
    _cp->put(CONN_POOL_TYPE_HBASE, hconn);
    return ret;
}

HResult DataOperations::ReadDirectoryInfo(const std::string& tablename, const std::string& rowkey, OUT DIR_INFO& dirinfo)
{
    HResult ret = ERROR_CODE_OK;
    
    // get connection
    hbase_conn *hconn = (hbase_conn *)_cp->get(CONN_POOL_TYPE_HBASE);
	HbaseOp *conn = hconn->conn;
    RowResult *result;
    
	if (hconn == NULL) {
		DERROR("ReadFileInfo error: hbase server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        goto done;
	}
    
    result = conn->get(tablename.c_str(), rowkey.c_str());
    
    if (result == NULL)
    {
        ret = conn->errnum;
        DERROR("hbase read row <%s> from table <%s> failed, error:%d\n", rowkey.c_str(), tablename.c_str(), ret);
        _cp->put(CONN_POOL_TYPE_HBASE, hconn);
        goto done;
    }
    
    ret = ReadFromRowResult(result, dirinfo);
    
    result->free();
    delete(result);
    
done:
    if (ret != ERROR_CODE_OK)
    {
        DERROR("Read directory info failed (error: %d): %s:%s", ret, tablename.c_str(), rowkey.c_str());
    }

    _cp->put(CONN_POOL_TYPE_HBASE, hconn);
    return ret;
}

HResult DataOperations::ListDirectoryInfo(const std::string& tablename, const std::string& startrow, OUT std::vector<DIR_INFO>& dirs)
{
    HResult ret = ERROR_CODE_OK;
    
    dirs.resize(0);
    
    // get connection
    hbase_conn *hconn = (hbase_conn *)_cp->get(CONN_POOL_TYPE_HBASE);
	HbaseOp *conn = NULL;
    ColumnValue col;
    int scancount = 10;
    col.column = DIR_COLUMNFAMILY;
    
    if (hconn == NULL) {
		DERROR("ListDirectoryInfo error: hbase server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        goto done;
	}
    
    conn = hconn->conn;
    
    ret = conn->scanopen(tablename.c_str(), startrow.c_str(), &col);
    if (ret != ERROR_CODE_OK)
    {
        DERROR("open scan for list directory failed: table:%s, startrow:%s, error:%d",
               tablename.c_str(), startrow.c_str(), ret);
        goto done;
    }
    
    while (ret == ERROR_CODE_OK) {
        RowResult *rows;
        
        rows = conn->scan(scancount);
        if (NULL == rows) {
            DINFO("can't get data in hbase\n");
            break;
        }
        DIR_INFO dirinfo;
        RowResult* row = rows;
        while (row)
        {
            if (strncmp(startrow.c_str(), row->row.c_str(), startrow.size()) != 0)
            {
                //stop the scan
                DNOTE("list directory finished: dircount: %d, start: %s, currentkey:%s", 
                      (int)dirs.size(), startrow.c_str(), row->row.c_str());
                ret = ERROR_CODE_FINISHED;
                break;
            }
            
            ret = ReadFromRowResult(row, dirinfo);
            DINFO("get data from row result, rowkey:%s, name:%s", row->row.c_str(), dirinfo._name);

            if (ret == ERROR_CODE_OK)
            {
                dirs.push_back(dirinfo);
            }
            else
            {
                DERROR("get data from row result failed: rowkey:%s",
                       row->row.c_str());
                break;
            }
            row = row->next;
        }
                       
        free_rows(rows);
    }
    
    if (ret == ERROR_CODE_FINISHED)
    {
        ret = ERROR_CODE_OK;
    }
    
done:
    _cp->put(CONN_POOL_TYPE_HBASE, hconn);
    return ret;
}

HResult DataOperations::WriteFileInfo(const std::string& tablename, const FILE_INFO& meta)
{
    HResult ret = ERROR_CODE_OK;
    
    // generate RowResult from File Meta info
    string rowkey = meta.GetKeyWithVersion();
    
    RowResult* row = GenerateRowResult(meta);
    
    // get connection
    hbase_conn *hconn = (hbase_conn *)_cp->get(CONN_POOL_TYPE_HBASE);
	HbaseOp *conn = NULL;
    
    int trytimes = 0;
    
	if (hconn == NULL) {
		DERROR("WriteFileInfo error: hbase server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        goto done;
	}
    
    conn = hconn->conn;
    //put lines in hbase
	while (1) {
		ret = conn->put(tablename.c_str(), rowkey.c_str(), row);
		if (ret < 0) {
			if (conn->errnum == ERROR_CODE_IOERROR) {
				DERROR("hbase put table <%s> error, io error, ignore the request.\n", tablename.c_str());
				ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
				goto done;
			} else if (conn->errnum == ERROR_CODE_SOCKET_CLOSED) {
				trytimes += 1;
				set_inactive(hconn);
				if (trytimes > 3) {
					DERROR("hbase put table <%s> try %d times, ignore the request.\n", tablename.c_str(), trytimes-1);
					ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
					goto done;
				}
				DERROR("hbase put table <%s> error, socket broken, try again(%d).\n", tablename.c_str(), trytimes);
				//hconn->pool->active = 0;
				_cp->put(CONN_POOL_TYPE_HBASE, hconn);
				hconn = (struct hbase_conn *)_cp->get(CONN_POOL_TYPE_HBASE);
				if (hconn == NULL) {
					DERROR("hbase put table <%s> error, server resource unavailable.\n", tablename.c_str());
					ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
					goto done;
				}
				conn = hconn->conn;
				continue;
			} else if (conn->errnum == ERROR_CODE_ILLEGAL_ARGUMENT) {
				DERROR("hbase put table <%s> error, invalid parameter.\n", tablename.c_str());
				ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
				goto done;
			} else if (conn->errnum == ERROR_CODE_DATA_STATUS_UNKNOWN) {
				DERROR("hbase put table <%s> error, data status unknown.\n", tablename.c_str());
				ret = ERROR_CODE_DATA_STATUS_UNKNOWN;
				goto done;
			}
		}
		break;
	}
    
done:
    free_rows(row);
    _cp->put(CONN_POOL_TYPE_HBASE, hconn);    
    return ret;    
}

HResult DataOperations::ReadFileInfo(const std::string& tablename, const std::string& rowkey, OUT FILE_INFO& fileinfo)
{
    HResult ret = ERROR_CODE_OK;
    
    // get connection
    hbase_conn *hconn = (hbase_conn *)_cp->get(CONN_POOL_TYPE_HBASE);
	HbaseOp *conn = hconn->conn;
    RowResult *result;
    
	if (hconn == NULL) {
		DERROR("ReadFileInfo error: hbase server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        goto done;
	}

    result = conn->get(tablename.c_str(), rowkey.c_str());
    
    if (result == NULL)
    {
        ret = conn->errnum;
        DERROR("hbase read row <%s> from table <%s> failed, error:%d", rowkey.c_str(), tablename.c_str(), ret);
        _cp->put(CONN_POOL_TYPE_HBASE, hconn);
        goto done;
    }
    
    ret = ReadFromRowResult(result, fileinfo);
    
    result->free();
    delete(result);
    
done: 
    if (ret != ERROR_CODE_OK)
    {
        DERROR("Read directory info failed (error: %d): %s:%s", ret, tablename.c_str(), rowkey.c_str());
    }
    
    _cp->put(CONN_POOL_TYPE_HBASE, hconn);
    return ret;
}

HResult DataOperations::ListFileInfo(const std::string& tablename, const std::string& startrow, OUT std::vector<FILE_INFO>& files)
{
    HResult ret = ERROR_CODE_OK;
    
    files.resize(0);
    
    // get connection
    hbase_conn *hconn = (hbase_conn *)_cp->get(CONN_POOL_TYPE_HBASE);
	HbaseOp *conn = NULL;
    int scancount = 10;
    ColumnValue col;
    col.column = FILE_COLUMNFAMILY;
    
    if (hconn == NULL) {
		DERROR("ListFileInfo error: hbase server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        goto done;
	}
    
    conn = hconn->conn;
    
    DINFO("open scan for list file: table:%s, startrow:%s",
          tablename.c_str(), startrow.c_str());
    
    ret = conn->scanopen(tablename.c_str(), startrow.c_str(), &col);
    if (ret != ERROR_CODE_OK)
    {
        DERROR("open scan for list file failed: table:%s, startrow:%s, error:%d",
               tablename.c_str(), startrow.c_str(), ret);
        goto done;
    }
    
    while (ret == ERROR_CODE_OK) {
        RowResult *rows;
        
        rows = conn->scan(scancount);
        if (NULL == rows) {
            DINFO("can't get data in hbase\n");
            break;
        }
        FILE_INFO fileinfo;
        RowResult* row = rows;
        while (row)
        {
            if (strncmp(startrow.c_str(), row->row.c_str(), startrow.size()) != 0)
            {
                //stop the scan
                DNOTE("list file finished: count: %d", (int)files.size());
                ret = ERROR_CODE_FINISHED;
                break;
            }
            
            DINFO("get data from row result, rowkey:%s", row->row.c_str());
            
            ret = ReadFromRowResult(row, fileinfo);
            if (ret == ERROR_CODE_OK)
            {
                files.push_back(fileinfo);
            }
            else
            {
                DERROR("get data from row result failed: rowkey:%s",
                       row->row.c_str());
                break;
            }
            row = row->next;
        }
        
        free_rows(rows);
    }
    
    if (ret == ERROR_CODE_FINISHED)
    {
        if (files.size() > 0)
        {
            ret = ERROR_CODE_OK;
        }
        else
        {
            ret = ERROR_CODE_EMPTY;
        }
    }
    
done:
    _cp->put(CONN_POOL_TYPE_HBASE, hconn);
    return ret;
}

HResult DataOperations::ListFileVersion(const std::string& tablename, const std::string& startrow, const std::string& filename, OUT std::vector<FILE_INFO>& files)
{
    HResult ret = ERROR_CODE_OK;
    
    files.resize(0);
    
    // get connection
    hbase_conn *hconn = (hbase_conn *)_cp->get(CONN_POOL_TYPE_HBASE);
	HbaseOp *conn = NULL;
    int scancount = 10;
    ColumnValue col;
    col.column = FILE_COLUMNFAMILY;
    
    if (hconn == NULL) {
		DERROR("ListFileInfo error: hbase server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        goto done;
	}
    
    conn = hconn->conn;
    
    DINFO("open scan for list file: table:%s, startrow:%s",
          tablename.c_str(), startrow.c_str());
    
    ret = conn->scanopen(tablename.c_str(), startrow.c_str(), &col);
    if (ret != ERROR_CODE_OK)
    {
        DERROR("open scan for list file failed: table:%s, startrow:%s, error:%d",
               tablename.c_str(), startrow.c_str(), ret);
        goto done;
    }
    
    while (ret == ERROR_CODE_OK) {
        RowResult *rows;
        
        rows = conn->scan(scancount);
        if (NULL == rows) {
            DINFO("can't get data in hbase\n");
            break;
        }
        FILE_INFO fileinfo;
        RowResult* row = rows;
        while (row)
        {
            if (strncmp(startrow.c_str(), row->row.c_str(), startrow.size()) != 0)
            {
                //stop the scan
                DNOTE("list file finished: count: %d", (int)files.size());
                ret = ERROR_CODE_FINISHED;
                break;
            }
            
            DINFO("get data from row result, rowkey:%s", row->row.c_str());
            
            ret = ReadFromRowResult(row, fileinfo);
            if (ret == ERROR_CODE_OK)
            {
                if (strcmp(fileinfo._name, filename.c_str()) == 0)
                {
                    files.push_back(fileinfo);
                }
            }
            else
            {
                DERROR("get data from row result failed: rowkey:%s",
                       row->row.c_str());
                break;
            }
            row = row->next;
        }
        
        free_rows(rows);
    }
    
    if (ret == ERROR_CODE_FINISHED)
    {
        if (files.size() > 0)
        {
            ret = ERROR_CODE_OK;
        }
        else
        {
            ret = ERROR_CODE_EMPTY;
        }
    }
    
done:
    _cp->put(CONN_POOL_TYPE_HBASE, hconn);
    return ret;
}

HResult DataOperations::WriteDataBlock(const std::string& filename, const vector<int8_t>& block, OUT DataLocation& location)
{
    HResult ret = ERROR_CODE_OK;
    
    uint64_t offset = 0;
    
    // get connection
    hdfs_conn *hconn = (hdfs_conn *)_cp->get(CONN_POOL_TYPE_HDFS);
    if (hconn == NULL) {
		DERROR("WriteDataBlock error: hdfs server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        return ret;
	}
    
	HdfsOp *conn = hconn->conn;
    
    HdfsHandle handle = NULL;
    
    if (conn->isExist(filename))
    {
        handle = conn->open(filename, FILE_MODE_APPEND);
    }
    else
    {
        handle = conn->open(filename, FILE_MODE_CREATE);
    }
    
    if (handle == NULL)
    {
        DERROR("hdfs can't open file:%s", filename.c_str());
        ret = ERROR_CODE_INTERNAL;
        goto done;
    }
    
    ret = conn->getlength(filename, offset);
    if (ret < 0)
    {
        DERROR("hdfs get file length failed :%s", filename.c_str());
        goto done;
    }
    
    ret = conn->writebytes(handle, block);
    if (ret < 0)
    {
        DERROR("hdfs write file failed :%s", filename.c_str());
        goto done;
    }
    
    uint64_t newlength;
    ret = conn->getlength(filename, newlength);
    if (ret < 0)
    {
        DERROR("hdfs get file length failed :%s", filename.c_str());
        goto done;
    }
    
    if (offset + block.size() != newlength)
    {
        DERROR("hdfs write file length is not valid :%s, start offset: %lu, size: %d, length after write: %lu",
               filename.c_str(), offset, (int)block.size(), newlength);
    }
    
    strcpy(location._fileName, filename.c_str());
    location._offset = offset;
    location._length = block.size();
    
done:
    if (handle != NULL)
    {
        conn->close(handle);
    }
    _cp->put(CONN_POOL_TYPE_HDFS, hconn);
    return ret;    
}

HResult DataOperations::ReadDataBlock(
        const std::string& filename, uint64_t offset, OUT int8_t* data, size_t length, OUT size_t& actualReadCount)
{
    HResult ret = ERROR_CODE_OK;
    DINFO("Read Data block, file: %s, offset: %lu, length: %u",
          filename.c_str(), offset, (uint32_t)length);
    // get connection
    hdfs_conn *hconn = (hdfs_conn *)_cp->get(CONN_POOL_TYPE_HDFS);
    if (hconn == NULL) {
		DERROR("ReadDataBlock error: hdfs server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        return ret;
	}
    
	HdfsOp *conn = hconn->conn;
    
    HdfsHandle handle = conn->open(filename, FILE_MODE_READ);
    if (handle == NULL)
    {
        DERROR("Open file failed, filename:%s", filename.c_str());
		ret = ERROR_CODE_IOERROR;
        _cp->put(CONN_POOL_TYPE_HDFS, hconn);
        return ret;
    }
    vector<int8_t> vec_data;
    ret = conn->readbytes(handle, offset, length, vec_data);

    if (ret == ERROR_CODE_OK)
    {
        actualReadCount = vec_data.size();
    
        if (length < actualReadCount)
        {
            // read more than request
            actualReadCount = length;
        }
    
        memcpy(data, &vec_data[0], actualReadCount);
    }
    else
    {
        DERROR("Read data failed, file=%s, offset=%lu, length=%u, err=%d",
               filename.c_str(), offset, (int)vec_data.size(), ret);
    }
    
    conn->close(handle);
    
    _cp->put(CONN_POOL_TYPE_HDFS, hconn);
    
    DINFO("Read Data block finished, file: %s, offset: %lu, length: %u, error: %d",
          filename.c_str(), offset, (uint32_t)actualReadCount, ret);
    return ret;
    
}

HResult DataOperations::GetNewDirId(USER_ID userid, DIR_ID& dirid)
{
    HResult ret = ERROR_CODE_OK;
    
    // get connection
    hdfs_conn *hconn = (hdfs_conn *)_cp->get(CONN_POOL_TYPE_HDFS);
    if (hconn == NULL)
    {
		DERROR("GetNewDirId error: hdfs server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        return ret;
	}
    
	HdfsOp *conn = hconn->conn;
    
    char filename[256];
    sprintf(filename, "%s/%u/%s", _rootdir, userid, DIRID_FILE);
    
    DIR_SN id;
    HdfsHandle handle;
    if (conn->isExist(filename))
    {
        handle = conn->open(filename, FILE_MODE_READ);
    
        vector<int8_t> vec_data;
        ret = conn->readbytes(handle, 0, sizeof(DIR_SN), vec_data);
    
        conn->close(handle);
    
        if (ret != ERROR_CODE_OK)
        {
            DERROR("Read dir id data file failed, file=%s, err=%d",
                   filename, ret);
        
            _cp->put(CONN_POOL_TYPE_HDFS, hconn);
            return ret;
        }
        DataStream stream(&vec_data);
        stream.reset();
        stream.UnSerialize(id);
        DINFO("Read DIR id out, %d", id);
        id ++;
    }
    else
    {
        // can't find the id file, will restart from root;
        id = ROOT_DIR_SN + 1;
    }
    
    // write back
    DataStream write;
    write.Serialize(id);
    
    handle = conn->open(filename, FILE_MODE_CREATE);
    if (handle == NULL)
    {
        DERROR("Open file failed, filename:%s", filename);
		ret = ERROR_CODE_IOERROR;
        goto done;
    }
    
    ret = conn->writebytes(handle, write.getdata());
    
    conn->close(handle);
    if (ret != ERROR_CODE_OK)
    {
        DERROR("Write dir id data file failed, file=%s, err=%d",
               filename, ret);
        goto done;
    }
    
    dirid._parts._dirSN = id;
    dirid._parts._userID = userid;
    
done:
    _cp->put(CONN_POOL_TYPE_HDFS, hconn);
    return ret;
}

// it will always create new files and let the output logic to decide
string DataOperations::GetNewFileName(USER_ID userid)
{
    char path[256];
    sprintf(path, "%s/%u/%s_%u", _rootdir, userid, DATA_FILE_PREFIX, (uint32_t)time(NULL));
    string filename(path);
    
    return filename;
}

HResult DataOperations::GetCurrentFileName(USER_ID userid, string& filename)
{
    HResult ret = ERROR_CODE_OK;
    
    // get connection
    hdfs_conn *hconn = (hdfs_conn *)_cp->get(CONN_POOL_TYPE_HDFS);
    if (hconn == NULL)
    {
		DERROR("GetCurrentFileName error: hdfs server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        return ret;
	}
    
    HdfsOp *conn = hconn->conn;
    char path[256];
    vector<FileStatus> files;
    sprintf(path, "%s/%u", _rootdir, userid);
    
    if (!conn->isExist(path)) // path does not exist
    {
        _cp->put(CONN_POOL_TYPE_HDFS, hconn);
        filename = GetNewFileName(userid);
        return ret;
    }
    
    ret = conn->listfiles(path, files);
    
    _cp->put(CONN_POOL_TYPE_HDFS, hconn);
    
    if (ret != ERROR_CODE_OK)
    {
        //TODO: handle NO more data to read
        DERROR("List File failed, path: %s, error: %d", path, ret);
        // treat as no data file
        ret = ERROR_CODE_OK;
    }
    
    uint32_t lastid = 0;
    for (size_t idx = 0; idx < files.size(); idx ++)
    {
        if (files[idx].isdir)
        {
            // skip directory
            continue;
        }
        DINFO("Find File:%s", files[idx].path.c_str());
        uint32_t id;
        size_t pos = files[idx].path.find(DATA_FILE_PREFIX);
        if (pos != string::npos)
        {
            string fileid = files[idx].path.substr(pos + strlen(DATA_FILE_PREFIX) + 1);
            DINFO("Find File ID:%s", fileid.c_str());
            id = atoi(fileid.c_str());
            
            if (id > lastid)
            {
                lastid = id;
                filename = files[idx].path;
            }
        }
    }

    if (lastid == 0)
    {
        filename = GetNewFileName(userid);
    }

    return ret;
}

HResult DataOperations::GetFileLength(const std::string& filename, uint64_t& length)
{
    HResult ret = ERROR_CODE_OK;
    
    // get connection
    hdfs_conn *hconn = (hdfs_conn *)_cp->get(CONN_POOL_TYPE_HDFS);
    if (hconn == NULL) {
		DERROR("ReadDataBlock error: hdfs server busy\n");
		ret = ERROR_CODE_SERVER_BUSY;
        return ret;
	}
    
    if (!hconn->conn->isExist(filename.c_str()))
    {
        ret = ERROR_CODE_EMPTY;
        _cp->put(CONN_POOL_TYPE_HDFS, hconn);
        return ret;
    }
    
    ret = hconn->conn->getlength(filename.c_str(), length);
    if (ret != ERROR_CODE_OK)
    {
        DERROR("Get File Length failed, name: %s, error: %d", filename.c_str(), ret);
    }
    
    _cp->put(CONN_POOL_TYPE_HDFS, hconn);
    return ret;
}


